package cn.itcast.up.model.statistisc

import cn.itcast.up.model.base.BaseModel
import org.apache.spark.sql.{Column, DataFrame}

/**
  * 年龄段标签计算
  */
object AgeModel extends BaseModel {

  import spark.implicits._
  import org.apache.spark.sql.functions._

  /**
    * 设置当前应用的名字
    *
    * @return
    */
  override def setAppName(): String = "AgeModel"

  /**
    * 设置4级标签ID
    *
    * @return
    */
  override def setFourTagID(): Int = 490

  /**
    * 标签计算
    *
    * @param hbaseDF HBase数据源
    * @param fiveDF  规则数据
    * @return 最终的计算结果:userid, tagIds
    */
  override def computeTagIds(hbaseDF: DataFrame, fiveDF: DataFrame): DataFrame = {
    //    hbaseDF.show()
    //    +---+----------+
    //| id|  birthday|
    //+---+----------+
    //|  1|1992-05-31|
    //| 10|1980-10-13|
    //|100|1993-10-28|
    //|101|1996-08-18|
    //|102|1996-07-28|
    //    fiveDF.show()
    //    +---+-----------------+
    //| id|             rule|
    //+---+-----------------+
    //|491|19500101-19591231|
    //|492|19600101-19691231|
    //|493|19700101-19791231|
    //|494|19800101-19891231|
    //|495|19900101-19991231|
    //|496|20000101-20091231|
    //|497|20100101-20191231|
    //+---+-----------------+

    //fiveDF 需要19500101-19591231 中间的-去掉
    val timeDF: DataFrame = fiveDF.map(row => {
      val tagIds: String = row.getAs[Long]("id").toString
      val rule: String = row.getAs[String]("rule").toString
      //对rule进行分割
      val arr: Array[String] = rule.split("-")
      (tagIds, arr(0), arr(1))
    }).collect().toList.toDF("tagIds", "start", "end")
    //    timeDF.show()
    //    +------+--------+--------+
    //|tagIds|   start|     end|
    //+------+--------+--------+
    //|   491|19500101|19591231|
    //|   492|19600101|19691231|
    //|   493|19700101|19791231|


    //hbaseDF 需要将1992-05-31 中间-去掉,
    val newBirthday: Column = regexp_replace(hbaseDF.col("birthday"), "-", "")
    val sourceDF: DataFrame = hbaseDF.select('id.as("userid"), newBirthday.as("birthday"))
    //      .show()
    //+------+--------+
    //|userid|birthday|
    //+------+--------+
    //|     1|19920531|
    //|    10|19801013|
    //|   100|19931028|
    //|   101|19960818|


    //使用 timeDF和sourceDF进行join关联,如果当前生日在某个范围之内,获取对应的标签ID
    sourceDF.join(timeDF)
      .where('birthday.between('start, 'end))
      //      .show()
      //+------+--------+------+--------+--------+
      //|userid|birthday|tagIds|   start|     end|
      //+------+--------+------+--------+--------+
      //|     1|19920531|   495|19900101|19991231|
      //|    10|19801013|   494|19800101|19891231|
      //|   100|19931028|   495|19900101|19991231|
      //|   101|19960818|   495|19900101|19991231|
      .select('userid, 'tagIds)
  }


  def main(args: Array[String]): Unit = {
//        executeCompute()
    //加载MySQL数据源
    val mysqlSource: DataFrame = getMysqlSource
    //获取4级规则的map
    val map: Map[String, String] = getFourRule(mysqlSource)
    //获取5级规则的DataFrame
    val fiveDF: DataFrame = getFiveDF(mysqlSource)
    //加载MySQL数据源
    val hbaseSource: DataFrame = loadHBaseData(map)
    //开始进行标签计算
    val newDF: DataFrame = computeTagIds(hbaseSource, fiveDF)
    //加载历史数据
    val oldDF: DataFrame = loadOldData(map)
    //进行标签合并
    val result: DataFrame = mergeTotalTag(newDF, oldDF)
    //将合并之后的数据落地
    sinkData(result, map)
  }
}
